package com.qdu.hyy.MovieHighestRating;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.util.Map;


import java.io.IOException;
import java.util.TreeMap;

public class MovieHighestRatingReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {

    TreeMap<DoubleWritable, String> treeMap;

    protected void setup(Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
        treeMap = new TreeMap<>();
    }

    protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws java.io.IOException, InterruptedException {


        Double value=new Double(0.0);
        for (DoubleWritable val :values)
        {


            //context.write(key, new DoubleWritable(sum));
            value=val.get();

        }
        treeMap.put(new DoubleWritable(value), key.toString());
        if (treeMap.size() > 10) {
            treeMap.remove(treeMap.firstKey());
        }
        System.out.println(key+"   "+values);



    }

    protected void cleanup(Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
        //降序输出

        for (Map.Entry<DoubleWritable, String> entry : treeMap.descendingMap().entrySet()) {
            context.write(new Text(entry.getValue()), entry.getKey());
        }

    }
}